package com.example.nacosdemo.consumer;

import com.alibaba.fastjson2.JSON;
import com.example.nacosdemo.dto.ReduceDTO;
import com.example.nacosdemo.service.ProductService;
import com.example.nacosdemo.service.UserService;
import lombok.AllArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.apache.kafka.clients.consumer.KafkaConsumer;
import org.springframework.context.annotation.Configuration;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.kafka.support.Acknowledgment;

@Configuration
@Slf4j
@AllArgsConstructor
public class KafkaConsumerConfig {

    private final UserService userService;

    private final ProductService productService;


    @KafkaListener(topics = "demo")
    public void demoConsumer(String message, Acknowledgment ack){
        log.info("message:{}", message);
        try {
            userService.kafkaAdd(message);
            ack.acknowledge();
        }catch (Exception e){
            e.printStackTrace();
        }
        log.info("消费成功");
    }

    @KafkaListener(topics = "reduce")
    public void reduce(String message, Acknowledgment ack){
        log.info("message:{}", message);
        ReduceDTO dto = JSON.parseObject(message, ReduceDTO.class);
        try {
            productService.reduceAndChangeOrder(dto);
            ack.acknowledge();
        }catch (Exception e){
            e.printStackTrace();
        }
        log.info("消费成功");
    }
}
